package com.spark.lola.consumer;

import com.alibaba.fastjson.JSON;
import com.spark.lola.entity.statistics.StatisticsStoreProduct;
import com.spark.lola.entity.warehouse.OrderInfo;
import com.spark.lola.service.*;
import com.spark.lola.support.common.constant.BooleanConstant;
import com.spark.lola.support.netty.common.NettyCacheUtils;
import com.spark.lola.support.netty.service.HawkPushServiceApi;
import com.spark.lola.support.warehouse.configuration.WarehouseKafkaTopic;
import com.spark.lola.support.wms.constant.WmsNettyCommands;
import io.netty.channel.Channel;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Set;

/**
 * @Author: DengYeJun
 * @Date: 2020/6/15 13:19
 */
@Service
public class CalConsumer {
    private final Logger logger = LoggerFactory.getLogger(getClass());
    @Autowired
    private IStatisticsService statisticsService;
    @Autowired
    private HawkPushServiceApi pushServiceApi;

    /**
     * 仓库 统计消费
     * @param consumerRecord
     */
    @KafkaListener(topics = WarehouseKafkaTopic.CAL_STORE)
    @Transactional(rollbackFor = Exception.class)
    public void calStore(ConsumerRecord<String, String> consumerRecord) throws Exception {
        Long tsp = System.currentTimeMillis();
        logger.info("===consumer【CAL_STORE】仓库【{}】参数【{}】===START===", consumerRecord.key(), consumerRecord.value());
        OrderInfo orderInfo = JSON.parseObject(consumerRecord.value(), OrderInfo.class);
        logger.info("更新仓库统计...");
        statisticsService.calStore(orderInfo);
        if (orderInfo.getStatus().equals(BooleanConstant.TRUE)) {
            logger.info("===订单状态不匹配,跳过 品种类目/仓库货品统计...");
            return;
        }
        logger.info("更新仓库货品统计...");
        StatisticsStoreProduct statistics = statisticsService.calStoreProduct(orderInfo);
        logger.info("===consumer【CAL_STORE】仓库【{}】参数【{}】耗时【{}】===END===", consumerRecord.key(), consumerRecord.value(),tsp/1000);
        Set<Channel> channels = NettyCacheUtils.getAllChannels();
        if(channels.size()>0) {
            pushServiceApi.pushMsg(channels, WmsNettyCommands.PUSH_WAREHOUSE_SOTRE_PRODUCT, JSON.toJSONString(statistics));
        }
    }

    /**
     * 货品类目 统计消费
     * @param consumerRecord
     */
    @KafkaListener(topics = WarehouseKafkaTopic.CAL_PRODUCT)
    @Transactional(rollbackFor = Exception.class)
    public void calProduct(ConsumerRecord<String, String> consumerRecord) throws Exception {
        Long tsp = System.currentTimeMillis();
        logger.info("===consumer【CAL_PRODUCT】货品【{}】参数【{}】===START===", consumerRecord.key(), consumerRecord.value());
        OrderInfo orderInfo = JSON.parseObject(consumerRecord.value(), OrderInfo.class);
        logger.info("更新品种类目统计...");
        statisticsService.calProduct(orderInfo);
        logger.info("===consumer【CAL_PRODUCT】货品【{}】参数【{}】耗时【{}】===END===", consumerRecord.key(), consumerRecord.value(),tsp/1000);
    }

    /**
     * 供应商 统计消费
     * @param consumerRecord
     */
    @KafkaListener(topics = WarehouseKafkaTopic.CAL_SUPPLIER)
    @Transactional(rollbackFor = Exception.class)
    public void calSupplier(ConsumerRecord<String, String> consumerRecord) throws Exception {
        Long tsp = System.currentTimeMillis();
        logger.info("===consumer【CAL_SUPPLIER】供应商【{}】参数【{}】===START===", consumerRecord.key(), consumerRecord.value());
        OrderInfo orderInfo = JSON.parseObject(consumerRecord.value(), OrderInfo.class);
        logger.info("更新供应商统计...");
        statisticsService.calSupplier(orderInfo);
        logger.info("===consumer【CAL_SUPPLIER】供应商【{}】参数【{}】耗时【{}】===END===", consumerRecord.key(), consumerRecord.value(),tsp/1000);
    }

}
